//! 文件存储的服务发现服务实现

use async_trait::async_trait;
use nacos_core::{ServiceStorage, ServiceInstance, Result, Storage};
use std::sync::Arc;

use super::storage::FileStorage;

#[derive(Debug)]
pub struct FileServiceStorage {
    fs: Arc<FileStorage>,
}

impl FileServiceStorage {
    pub fn new(fs: Arc<FileStorage>) -> Self {
        Self { fs }
    }

    fn get_key(namespace: &str, group: &str, service_name: &str, instance_id: &str) -> String {
        format!("service/{}@@{}@@{}@@{}", namespace, group, service_name, instance_id)
    }
}

#[async_trait]
impl ServiceStorage for FileServiceStorage {
    async fn register_instance(&self, instance: &ServiceInstance) -> Result<()> {
        let key = Self::get_key(&instance.namespace, &instance.group_name, &instance.service_name, &instance.instance_id);
        let bytes = serde_json::to_vec(instance)?;
        self.fs.put(&key, &bytes).await
    }

    async fn deregister_instance(&self, namespace: &str, group: &str, service_name: &str, instance_id: &str) -> Result<()> {
        let key = Self::get_key(namespace, group, service_name, instance_id);
        self.fs.delete(&key).await
    }

    async fn get_instance(&self, namespace: &str, group: &str, service_name: &str, instance_id: &str) -> Result<Option<ServiceInstance>> {
        let key = Self::get_key(namespace, group, service_name, instance_id);
        if let Some(bytes) = self.fs.get(&key).await? {
            Ok(serde_json::from_slice(&bytes).ok())
        } else {
            Ok(None)
        }
    }

    async fn get_all_instances(&self, namespace: &str, group: &str, service_name: &str) -> Result<Vec<ServiceInstance>> {
        let prefix = format!("service/{}@@{}@@{}@@", namespace, group, service_name);
        let all_items = self.fs.scan(&prefix).await?;
        let result = all_items.into_iter()
            .filter_map(|(_, v)| serde_json::from_slice::<ServiceInstance>(&v).ok())
            .collect();
        Ok(result)
    }

    async fn update_instance_health(&self, namespace: &str, group: &str, service_name: &str, instance_id: &str, healthy: bool) -> Result<()> {
        if let Some(mut instance) = self.get_instance(namespace, group, service_name, instance_id).await? {
            instance.healthy = healthy;
            self.register_instance(&instance).await?;
        }
        Ok(())
    }

    async fn get_all_services(&self, namespace: &str, group: Option<&str>) -> Result<Vec<String>> {
        let prefix = format!("service/{}@@", namespace);
        let all_items = self.fs.scan(&prefix).await?;
        let services = all_items.into_iter()
            .filter_map(|(_, v)| serde_json::from_slice::<ServiceInstance>(&v).ok())
            .filter(|i| group.map_or(true, |g| i.group_name == g))
            .map(|i| i.service_name)
            .collect::<std::collections::HashSet<String>>()
            .into_iter()
            .collect();
        Ok(services)
    }

    async fn scan(&self, prefix: &str) -> Result<Vec<(String, Vec<u8>)>> {
        self.fs.scan(prefix).await
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use nacos_core::ServiceInstance;
    use tempfile::tempdir;

    async fn create_test_storage() -> FileServiceStorage {
        let dir = tempdir().unwrap();
        let file_storage = Arc::new(FileStorage::new(dir.path()).await.unwrap());
        FileServiceStorage::new(file_storage)
    }

    fn create_test_instance(service_name: &str, ip: &str, port: u16) -> ServiceInstance {
        let mut instance = ServiceInstance::new(service_name.to_string(), "test-group".to_string(), ip.to_string(), port);
        instance.namespace = "public".to_string();
        instance
    }

    #[tokio::test]
    async fn test_register_and_get_instance() {
        let storage = create_test_storage().await;
        let instance = create_test_instance("test-service", "127.0.0.1", 8080);

        storage.register_instance(&instance).await.unwrap();

        let retrieved = storage.get_instance("public", "test-group", "test-service", &instance.instance_id).await.unwrap().unwrap();
        assert_eq!(instance, retrieved);
    }

    #[tokio::test]
    async fn test_deregister_instance() {
        let storage = create_test_storage().await;
        let instance = create_test_instance("test-service", "127.0.0.1", 8080);
        storage.register_instance(&instance).await.unwrap();

        storage.deregister_instance("public", "test-group", "test-service", &instance.instance_id).await.unwrap();

        let retrieved = storage.get_instance("public", "test-group", "test-service", &instance.instance_id).await.unwrap();
        assert!(retrieved.is_none());
    }

    #[tokio::test]
    async fn test_get_all_instances() {
        let storage = create_test_storage().await;
        storage.register_instance(&create_test_instance("service-a", "1.1.1.1", 80)).await.unwrap();
        storage.register_instance(&create_test_instance("service-a", "2.2.2.2", 80)).await.unwrap();
        storage.register_instance(&create_test_instance("service-b", "3.3.3.3", 80)).await.unwrap();

        let instances = storage.get_all_instances("public", "test-group", "service-a").await.unwrap();
        assert_eq!(instances.len(), 2);
    }

    #[tokio::test]
    async fn test_update_instance_health() {
        let storage = create_test_storage().await;
        let instance = create_test_instance("health-service", "4.4.4.4", 80);
        storage.register_instance(&instance).await.unwrap();

        storage.update_instance_health("public", "test-group", "health-service", &instance.instance_id, false).await.unwrap();

        let retrieved = storage.get_instance("public", "test-group", "health-service", &instance.instance_id).await.unwrap().unwrap();
        assert!(!retrieved.healthy);
    }

    #[tokio::test]
    async fn test_get_all_services() {
        let storage = create_test_storage().await;
        storage.register_instance(&create_test_instance("service-x", "1.1.1.1", 80)).await.unwrap();
        storage.register_instance(&create_test_instance("service-y", "2.2.2.2", 80)).await.unwrap();
        let mut other_group = create_test_instance("service-z", "3.3.3.3", 80);
        other_group.group_name = "other-group".to_string();
        storage.register_instance(&other_group).await.unwrap();

        let services = storage.get_all_services("public", Some("test-group")).await.unwrap();
        assert_eq!(services.len(), 2);
        assert!(services.contains(&"service-x".to_string()));
        assert!(services.contains(&"service-y".to_string()));

        let all_services = storage.get_all_services("public", None).await.unwrap();
        assert_eq!(all_services.len(), 3);
    }
}